package com.iot.processor.consumer;

import com.iot.processor.model.SensorData;
import com.iot.processor.parser.MessageParser;
import com.iot.processor.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.LoggerFactory;


import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;


/**
 * 数据消费者
 * 从消息队列中消费消息，解析并存储到Redis
 */
@Slf4j
public class DataConsumer implements Runnable {


    
    private final BlockingQueue<String> messageQueue;
    private final RedisService redisService;
    private final MessageParser messageParser;
    private final AtomicBoolean running;
    private final String consumerName;
    
    // 统计信息
    private final AtomicLong processedCount;
    private final AtomicLong successCount;
    private final AtomicLong errorCount;
    
    // 配置参数
    private static final int POLL_TIMEOUT_SECONDS = 1;  // 队列轮询超时时间
    private static final int BATCH_SIZE = 10;           // 批处理大小
    
    /**
     * 构造函数
     * 
     * @param messageQueue 消息队列
     * @param redisService Redis服务
     * @param consumerName 消费者名称
     */
    public DataConsumer(BlockingQueue<String> messageQueue, RedisService redisService, String consumerName) {
        this.messageQueue = messageQueue;
        this.redisService = redisService;
        this.consumerName = consumerName;
        this.messageParser = new MessageParser();
        this.running = new AtomicBoolean(false);
        this.processedCount = new AtomicLong(0);
        this.successCount = new AtomicLong(0);
        this.errorCount = new AtomicLong(0);
    }
    
    @Override
    public void run() {
        running.set(true);
        log.info("数据消费者 {} 开始运行", consumerName);
        
        while (running.get()) {
            try {
                // 从队列中获取消息，带超时机制
                String message = messageQueue.poll(POLL_TIMEOUT_SECONDS, TimeUnit.SECONDS);
                
                if (message != null) {
                    processMessage(message);
                } else {
                    // 队列为空时的处理逻辑
                    handleEmptyQueue();
                }
                
            } catch (InterruptedException e) {
                log.info("数据消费者 {} 被中断", consumerName);
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("数据消费者 {} 发生未知异常", consumerName, e);
                errorCount.incrementAndGet();
            }
        }

        log.info("数据消费者 {} 停止运行，处理统计 - 总数: {}, 成功: {}, 失败: {}",
                   consumerName, processedCount.get(), successCount.get(), errorCount.get());
    }
    
    /**
     * 处理单条消息
     * 
     * @param message 十六进制消息字符串
     */
    private void processMessage(String message) {
        processedCount.incrementAndGet();
        
        try {
            // 解析消息
            SensorData sensorData = messageParser.parseMessage(message);
            
            // 验证数据合理性
            if (!messageParser.validateSensorData(sensorData)) {
                log.warn("传感器数据验证失败，跳过存储: {}", message);
                errorCount.incrementAndGet();
                return;
            }
            
            // 存储到Redis
            boolean stored = redisService.storeSensorData(sensorData);
            
            if (stored) {
                successCount.incrementAndGet();
                log.debug("消费者 {} 成功处理消息: {} -> {}",
                           consumerName, message, sensorData.getDeviceId());
            } else {
                errorCount.incrementAndGet();
                log.warn("消费者 {} 存储数据失败: {}", consumerName, message);
            }
            
        } catch (IllegalArgumentException e) {
            errorCount.incrementAndGet();
            log.info("消费者 {} 解析消息失败: {} - {}", consumerName, message, e.getMessage());
        } catch (Exception e) {
            errorCount.incrementAndGet();
            log.info("消费者 {} 处理消息时发生异常: {}", consumerName, message, e);
        }
    }
    
    /**
     * 批量处理消息（可选的优化方案）
     */
    private void processBatchMessages() {
        SensorData[] batch = new SensorData[BATCH_SIZE];
        int batchIndex = 0;
        
        try {
            // 收集一批消息
            while (batchIndex < BATCH_SIZE && running.get()) {
                String message = messageQueue.poll(100, TimeUnit.MILLISECONDS);
                if (message == null) {
                    break; // 没有更多消息，处理当前批次
                }
                
                try {
                    SensorData sensorData = messageParser.parseMessage(message);
                    if (messageParser.validateSensorData(sensorData)) {
                        batch[batchIndex++] = sensorData;
                    }
                    processedCount.incrementAndGet();
                } catch (Exception e) {
                    errorCount.incrementAndGet();
                    log.error("批处理中解析消息失败: {}", message, e);
                }
            }
            
            // 批量存储
            if (batchIndex > 0) {
                SensorData[] actualBatch = new SensorData[batchIndex];
                System.arraycopy(batch, 0, actualBatch, 0, batchIndex);
                
                int storedCount = redisService.batchStoreSensorData(actualBatch);
                successCount.addAndGet(storedCount);
                errorCount.addAndGet(batchIndex - storedCount);

                log.debug("消费者 {} 批量处理完成: 处理 {} 条，成功存储 {} 条",
                           consumerName, batchIndex, storedCount);
            }
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            log.error("消费者 {} 批量处理时发生异常", consumerName, e);
        }
    }
    
    /**
     * 处理队列为空的情况
     */
    private void handleEmptyQueue() {
        // 队列为空时可以进行一些维护工作
        if (processedCount.get() % 1000 == 0 && processedCount.get() > 0) {
            log.info("消费者 {} 处理进度 - 总数: {}, 成功: {}, 失败: {}, 队列大小: {}",
                       consumerName, processedCount.get(), successCount.get(), 
                       errorCount.get(), messageQueue.size());
        }
    }
    
    /**
     * 停止消费者
     */
    public void stop() {
        running.set(false);
        log.info("数据消费者 {} 收到停止信号", consumerName);
    }
    
    /**
     * 检查消费者是否正在运行
     */
    public boolean isRunning() {
        return running.get();
    }
    
    /**
     * 获取处理统计信息
     */
    public ProcessingStats getStats() {
        return new ProcessingStats(
            consumerName,
            processedCount.get(),
            successCount.get(),
            errorCount.get(),
            running.get()
        );
    }
    
    /**
     * 获取消费者状态信息
     */
    public String getStatus() {
        double successRate = processedCount.get() > 0 ? 
            (double) successCount.get() / processedCount.get() * 100 : 0;
        
        return String.format("消费者 %s - 运行状态: %s, 处理总数: %d, 成功: %d, 失败: %d, 成功率: %.2f%%, 队列大小: %d", 
                           consumerName, running.get() ? "运行中" : "已停止", 
                           processedCount.get(), successCount.get(), errorCount.get(), 
                           successRate, messageQueue.size());
    }
    
    /**
     * 处理统计信息内部类
     */
    public static class ProcessingStats {
        private final String consumerName;
        private final long processedCount;
        private final long successCount;
        private final long errorCount;
        private final boolean running;
        
        public ProcessingStats(String consumerName, long processedCount, 
                             long successCount, long errorCount, boolean running) {
            this.consumerName = consumerName;
            this.processedCount = processedCount;
            this.successCount = successCount;
            this.errorCount = errorCount;
            this.running = running;
        }
        
        public String getConsumerName() { return consumerName; }
        public long getProcessedCount() { return processedCount; }
        public long getSuccessCount() { return successCount; }
        public long getErrorCount() { return errorCount; }
        public boolean isRunning() { return running; }
        
        public double getSuccessRate() {
            return processedCount > 0 ? (double) successCount / processedCount * 100 : 0;
        }
        
        @Override
        public String toString() {
            return String.format("ProcessingStats{消费者='%s', 处理总数=%d, 成功=%d, 失败=%d, 成功率=%.2f%%, 运行中=%s}",
                               consumerName, processedCount, successCount, errorCount, getSuccessRate(), running);
        }
    }
}